package com.gzhryc.shared_device.oem.micro.api;

import com.gzhryc.common.event.DelayHandleService;
import com.gzhryc.common.json.JsonTools;
import com.gzhryc.common.kafka.consumer.interfaces.KafkaConsumerEventListener;
import com.gzhryc.common.kafka.models.ConsumerEvent;
import com.gzhryc.common.logging.Logger;
import com.gzhryc.jdbc.JdbcSession;
import com.gzhryc.shared_device.oem.code.KafkaEventConstants;
import com.gzhryc.shared_device.oem.code.order.event.OrderCreateEvent;

import org.apache.commons.lang3.StringUtils;

import java.util.List;

public class ConfigKafkaEventListener implements KafkaConsumerEventListener {

	final static Logger log = Logger.getLogger(ConfigKafkaEventListener.class);

	String topic;

	public ConfigKafkaEventListener() {}

	@Override
	public void start(String topic) {
		this.topic = topic;
		log.info("开始监听主题：" + topic);
	}

	@Override
	public void receive(List<ConsumerEvent> events) {
		for (ConsumerEvent event : events) {
			String json = event.getValue();
			log.info("主题：{{0}}，组：{{1}}，区：{{2}}，获取消息，KEY={{3}}，VALUE={{4}}", event.getTopic(), event.getGroupId(),
					event.getPartition(), event.getKey(), event.getValue());
			if (StringUtils.isNotBlank(json)) {
				try {
					if (KafkaEventConstants.KEY_PLACE_GPS.equals(event.getKey())) {
						OrderCreateEvent businessEvent = JsonTools.fromJson(json, OrderCreateEvent.class);
						DelayHandleService.get().addEvent(businessEvent);
					}
				}catch (Exception e) {
					log.error("解析配置变更事件错误", e);
				} finally {
					JdbcSession.clear();
				}
			}
		}
	}

	@Override
	public void close(String topic) {
		log.info("结束监听主题：" + topic);
	}
}
